import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MasterService {

  protected static final Logger log = LoggerFactory.getLogger(MasterService.class);

  private CuratorFramework client;
  public static final String ROOT = "/gunode";
  public static final String MASTER_NODE = ROOT + "/master";

  public MasterService() {
    client = CuratorFrameworkFactory.builder().connectString("gcb.aliyun.com:2181")
        .retryPolicy(new RetryNTimes(1, 1000))
        .build();
    client.start();

    try {
      initRoot();
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  private void initRoot() throws Exception {
    String rootPath = ROOT;
    Stat stat = client.checkExists().forPath(rootPath);
    if (stat == null) {
      log.info("init root path {}", rootPath);
      client.create().forPath(rootPath);
    }
  }

  public Node electMaster(Node node, MasterListener listener) throws Exception {
    try {
      client.create().withMode(CreateMode.EPHEMERAL).forPath(MASTER_NODE, node.getId().getBytes());
      node.setMaster(true);
      log.info("i am master {}", node.getId());
      return node;
    } catch (Exception e) {
      if (e instanceof KeeperException) {
        KeeperException ke = (KeeperException) e;
        if (ke.code().equals(Code.NODEEXISTS)) {
          node.setMaster(false);
          byte[] bytes = client.getData().forPath(MASTER_NODE);
          String id = new String(bytes);
          log.info("exist master {}", id);
          // 订阅
          subscribe(MASTER_NODE, node, listener);
          return new Node(id, true);
        } else {
          throw e;
        }
      } else {
        throw e;
      }
    }
  }

  public void subscribe(String masterNode, Node node,
      MasterListener listener) throws Exception {
    System.out.println("subscribe master");
    client.getChildren().usingWatcher((CuratorWatcher) event -> listener.whenDown(node))
        .forPath(masterNode);
  }

  public void close() {
    try {
      if (client.checkExists().forPath(MASTER_NODE) != null) {
        client.delete().forPath(MASTER_NODE);
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

}
